
#include "master.h"
#include "slave.h"
#include "slave_protocol.h"
#include "../base/kvcfg.h"
#include "../base/mq.h"
#include "../base/list.h"
#include "../base/map.h"
#include "../base/th.h"
#include "../base/dl.h"

#include <assert.h>
#include <stdio.h>

C_CODE_BEGIN

/*
 * This command executor specifically load a slave into the memory 
 * and prepare all its function and configuration .
 */


const char* _cmd_add_notification_slave_name() {
	static const char* _name="master-notification";
	return _name;
}

const char* _cmd_load_slave_name() {
	static const char* _name="load-slave";
	return _name;
}

const char* _cmd_add_thread_name() {
	static const char* _name="add-thread";
	return _name;
}

static void _slave_notification_cb( struct message* msg , void* arg , struct slave_api* api ) {
	/* do not use _slave_api_get_master since it is the internal API for special purpose */
	struct master* ms = (struct master*)arg;
	/* notifying a certain slave */
	struct slave* sl = (struct slave*)(msg->msg);
	/* invoke the slave's notification function */
	_slave_notification_invoke(sl);
	/* enqueue the message queue to the global queue to enable back end thread 
	 * start to work for this message */
	msgqueue_enqueue(ms->glb_msgqueue,&sl);
	
}

int _cmd_add_notification_slave( struct master* ms , struct kv_element* val ) {
	struct slave_protocol spl;
	struct msgqueue* mq;
	struct slave* sl;
	struct slave_api api;
	struct map_kv mkv;
	spl.clean = NULL;
	spl.init = NULL;
	spl.argument = ms;
	spl.notify = NULL;
	spl.cb = _slave_notification_cb;
	api.sendmessage = _master_send_msg;
	/* create its internal message queue */
	mq = msgqueue_create(20,sizeof(struct message));
	sl = _slave_compose(&spl,mq,&api,ms);
	msgqueue_enqueue(ms->glb_msgqueue,&sl);
	mkv.key = _cmd_add_notification_slave_name();
	mkv.val = &sl;
	map_insert(ms->slavemap,&mkv,NULL);
	printf("================================================\n");
	printf("The internal slave %s is installed successfully!\n",_cmd_add_notification_slave_name());
	return 1;
}


static bool _cmd_load_slave_foreach( int idx, struct kv_element* v , void* arg ) {
	struct master* ms = (struct master*)arg;
	const char* dl_path = kv_element_get_atomic_value(v);
	struct msgqueue* mq;
	void* pvret;
	struct slave* sl;
	struct slave_api api;
	struct map_kv mkv;
	int iret;
	struct message notify_msg;
	api.sendmessage = _master_send_msg;

	assert(kv_element_get_type(v)==KV_ATOM);

	/* initialize the message queue for the slave */
	mq = msgqueue_create(20,sizeof(struct message));
	printf("================================================\n");
	printf("The slave whose dl path is %s is installing now!\n",dl_path);
	/* load the slave manually */
	sl = _slave_load( dl_path , mq ,&api,ms);
	if( sl == NULL ) {
		msgqueue_destroy(mq);
		fprintf(stderr,"Can not load slave whose dl path is %s!\n",dl_path);
	} else {
		iret= _slave_init_invoke(sl);
		if( iret <=0 ) {
			fprintf(stderr,"The initialization function of slave %s can not be invoked properly!\n",_slave_name(sl));
			/* destroy related resource */
			if( iret == 0 )
				_slave_unload(sl);
			/* destroy the message queue since this slave is loaded with error */
			msgqueue_destroy(mq);
		} else {
			/* insert the slave into the global map */
			mkv.key = _slave_name(sl);
			mkv.val = &sl;
			/* insert the slave pointer into the map */
			pvret = map_insert(ms->slavemap,&mkv,NULL);
			if( pvret == NULL ) {
				/* can not insert this slave */
				fprintf(stderr,"The slave %s can not be installed, due to the name \
							   collision or name length is over 64 characters!\n",_slave_name(sl));
				/* destroy all the related resource */
				_slave_unload(sl);
				msgqueue_destroy(mq);
			} else {
				/* add notification message which will be invoked later */
				notify_msg.msg = sl;
				notify_msg.clean=NULL;
				_master_send_message_by_slavename(ms,&notify_msg,_cmd_add_notification_slave_name());
				printf("The slave %s is installed successfully!\n",_slave_name(sl));
			}
		}
	}
	return true;
}

int _cmd_load_slave( struct master* ms , struct kv_element* val ) {
	if (val)
		kv_element_ls_foreach( val , _cmd_load_slave_foreach , ms );
	else {
		printf("================================================\n");
		printf("No slave is loaded!\n");
	}
	return 1;
}

static const char* _name_worker_thread() {
	static size_t count=0;
	static char buf[1024];
	++count;
	sprintf(buf,"master_worker_thread_%d",count);
	return buf;
}


static void* _thread_cb( void* arg ) {
	struct master* m = (struct master*)(arg);
	struct slave* sl_ptr;
	struct msgqueue* mq;
	struct message msg;

	while(1) {
		/* on idle */
		if( strcmp(m->thread_cmd,"EXIT") == 0 )
			break;
		/* process message in the global queue */
		if( msgqueue_xdequeue(m->glb_msgqueue,&sl_ptr) != NULL ) {
			/* get the message queue */
			mq = _slave_ref_msgqueue(sl_ptr);
			/* get a message from the queue */
			if( msgqueue_xdequeue(mq,&msg) != NULL ) {
				_slave_cb_invoke(sl_ptr,&msg);
				/* clean the message if the message wish */
				if( msg.clean )
					msg.clean(&msg);
			}
			/* push the queue back to the global queue */
			msgqueue_enqueue(m->glb_msgqueue,&sl_ptr);
		} else {
			/* let other thread comes in  */
			/* TODO:: add a better spin lock */
			thread_yield();
		}
	}
	return NULL;
}


int _cmd_add_thread( struct master* ms , struct kv_element* kv ) {
	/* config the thread number 
	 * The thread configuration should always put at the last the configuration process
	 * since the thread will run separately and the resource should be prepared well when
	 * configuring the thread of the master 
	 */
	int intval;
	int cnt;
	thread_handler th;
	const char* val;
	if( kv ) {
		val = kv_element_get_atomic_value(kv);
		assert( kv_element_get_type(kv) == KV_ATOM );
		intval = atoi(val);
		intval = intval <= 0 ? 2 : intval;
	} else {
		intval = 2;
	}

	for( cnt = 0 ; cnt < intval ; ++cnt ) {
		if( !thread_create(&th,_name_worker_thread(),_thread_cb,ms,false) ) {
			fprintf(stderr,"A master thread can not be created!\n");
		} else {
			list_push_back(ms->thpool,&th);
		}
	}
	printf("================================================\n");
	printf("%d master thread has been initialized.\n",intval);
	return 1;
}

C_CODE_END